package com.bestcem.xm.component.mqmetrics.service;

import com.bestcem.xm.component.mqmetrics.properties.MqMetricProperties;
import io.micrometer.core.instrument.ImmutableTag;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Tag;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

public class CustomMqMetricService implements InitializingBean {

    private MqMetricProperties properties;

    private MeterRegistry registry;

    @Autowired
    public void setProperties(MqMetricProperties properties) {
        this.properties = properties;
    }

    @Autowired
    public void setRegistry(MeterRegistry registry) {
        this.registry = registry;
    }


    private ConcurrentHashMap<String, Integer> successMap;
    private ConcurrentHashMap<String, Integer> errorMap;
    private ConcurrentHashMap<String, Integer> offsetMap;
    private List<Tag> commonTags;
    private String mqType;

    private void init() {
        mqType = ObjectUtils.isEmpty(properties.getMqType()) ? "rabbit" : properties.getMqType();
        String clientId = ObjectUtils.isEmpty(properties.getClientId()) ? "127.0.0.1" : properties.getClientId();
        successMap = new ConcurrentHashMap<>();
        errorMap = new ConcurrentHashMap<>();
        offsetMap = new ConcurrentHashMap<>();

        commonTags = new ArrayList<>();
        commonTags.add(new ImmutableTag("client_id", "java".concat(" ").concat(clientId)));
        commonTags.add(new ImmutableTag("mq_type", mqType));
    }

    public void gauge(String topic, Integer offset) {
        List<Tag> tags = new ArrayList<>(commonTags);
        tags.add(new ImmutableTag("topic", topic));
        registry.gauge("java_coroutine_change", tags, Double.NaN);
        registry.gauge("java_consumer_timeout_count", tags, Double.NaN);
        registry.gauge("java_consumer_retry_count", tags, Double.NaN);

        if (Objects.equals("kafka", mqType)) {
            offsetMap.put(topic, offset);
            registry.gauge("java_kafka_consume_offset", tags, offsetMap, p -> p.get(topic));
        }
        registry.gauge("java_consumer_success_count", tags, successMap, p -> p.get(topic));
        registry.gauge("java_consumer_failure_count", tags, errorMap, p -> p.get(topic));
    }

    public void metricSuccess(String topic, Integer offset) {
        Integer sc = CollectionUtils.isEmpty(successMap) ? 1 : (successMap.get(topic) == null ? 1 : successMap.get(topic) + 1);
        successMap.put(topic, sc);
        gauge(topic, offset);
    }

    public void metricError(String topic, Integer offset) {
        Integer ec = CollectionUtils.isEmpty(errorMap) ? 1 : (errorMap.get(topic) == null ? 1 : errorMap.get(topic) + 1);
        successMap.put(topic, ec);
        gauge(topic, offset);
    }

    @Override
    public void afterPropertiesSet() {
        init();
    }
}
